package FutureAndPromise.Demo04_NettyPromise;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Promise;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutionException;

/**
 * Demo04 - Netty ChannelPromise 详解 - 客户端
 * 
 * 学习目标：
 * 1. 理解 Promise 是可写的 Future
 * 2. 掌握如何创建和使用 Promise
 * 3. 学习 Promise 的典型使用场景
 * 4. 了解 Promise 和 Future 的关系
 */
@Slf4j
public class Client {

    public static void main(String[] args) throws Exception {
        log.info("=== Netty ChannelPromise 详解 - 客户端 ===\n");
        log.info("请先启动 Server.java\n");
        
        demo1_BasicPromise();
        demo2_ChannelPromise();
        demo3_BridgingAsync();
        demo4_PromiseWithRetry();
        demo5_TryMethods();
        
        Thread.sleep(1000);
        System.exit(0);
    }

    /**
     * Demo 1: Promise 基础概念
     */
    private static void demo1_BasicPromise() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 1: Promise 基础概念");
        log.info("└─────────────────────────────────────────\n");
        
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            // 1. 创建 Promise
            log.info("1. 创建 Promise（可写的 Future）");
            Promise<String> promise = new DefaultPromise<>(group.next());
            
            // 2. 作为 Future 使用（消费者视角）
            log.info("2. 添加监听器（作为 Future）");
            promise.addListener(future -> {
                if (future.isSuccess()) {
                    log.info("   ✅ [监听器] 操作成功，结果: {}", future.getNow());
                } else {
                    log.error("   ❌ [监听器] 操作失败: {}", future.cause().getMessage());
                }
            });
            
            // 3. 在另一个线程中设置结果（生产者视角）
            log.info("3. 在异步任务中设置结果（作为 Promise）");
            new Thread(() -> {
                try {
                    log.info("   [异步任务] 执行中...");
                    Thread.sleep(500);
                    
                    // 设置成功结果
                    promise.setSuccess("任务完成!");
                    log.info("   [异步任务] 已设置成功结果");
                    
                } catch (Exception e) {
                    // 设置失败原因
                    promise.setFailure(e);
                }
            }).start();
            
            // 4. 等待结果（可选）
            log.info("4. 等待异步结果...");
            String result = promise.sync().getNow();
            log.info("   最终结果: {}", result);
            
            Thread.sleep(200);
            
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * Demo 2: ChannelPromise 的使用
     */
    private static void demo2_ChannelPromise() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 2: ChannelPromise 的使用");
        log.info("└─────────────────────────────────────────\n");
        
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new SimpleHandler());
                    }
                });
            
            // 连接服务器
            ChannelFuture connectFuture = b.connect("localhost", 9000).sync();
            Channel channel = connectFuture.channel();
            
            // 1. 创建 ChannelPromise
            log.info("1. 创建 ChannelPromise");
            ChannelPromise promise = channel.newPromise();
            
            // 2. 添加监听器
            promise.addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.info("   ✅ Promise 操作成功");
                } else {
                    log.error("   ❌ Promise 操作失败: {}", future.cause().getMessage());
                }
            });
            
            // 3. 发送数据并设置 Promise 的结果
            log.info("2. 发送数据...");
            channel.writeAndFlush(
                Unpooled.copiedBuffer("Test\n", StandardCharsets.UTF_8)
            ).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.info("   数据发送成功，设置 Promise 为成功");
                    promise.setSuccess();
                } else {
                    log.error("   数据发送失败，设置 Promise 为失败");
                    promise.setFailure(future.cause());
                }
            });
            
            // 4. 等待 Promise 完成
            promise.sync();
            log.info("3. Promise 已完成");
            
            Thread.sleep(500);
            channel.close().sync();
            
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * Demo 3: 使用 Promise 桥接异步操作
     */
    private static void demo3_BridgingAsync() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 3: 使用 Promise 桥接异步操作");
        log.info("└─────────────────────────────────────────\n");
        
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new SimpleHandler());
                    }
                });
            
            ChannelFuture connectFuture = b.connect("localhost", 9000).sync();
            Channel channel = connectFuture.channel();
            
            // 封装一个返回 Promise 的方法
            log.info("调用自定义异步方法");
            ChannelFuture sendFuture = sendMessageAsync(channel, "Hello Promise\n");
            
            // 使用返回的 Future
            sendFuture.addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.info("✅ 自定义异步操作成功");
                } else {
                    log.error("❌ 自定义异步操作失败");
                }
            });
            
            sendFuture.sync();
            
            Thread.sleep(500);
            channel.close().sync();
            
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * 自定义异步方法：返回 Promise
     */
    private static ChannelFuture sendMessageAsync(Channel channel, String message) {
        log.info("   [异步方法] 准备发送消息: {}", message.trim());
        
        // 创建 Promise
        ChannelPromise promise = channel.newPromise();
        
        // 执行异步操作
        channel.eventLoop().execute(() -> {
            log.info("   [异步方法] 在 EventLoop 中执行");
            
            channel.writeAndFlush(
                Unpooled.copiedBuffer(message, StandardCharsets.UTF_8)
            ).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.info("   [异步方法] 消息发送成功，设置 Promise");
                    promise.setSuccess();
                } else {
                    log.error("   [异步方法] 消息发送失败");
                    promise.setFailure(future.cause());
                }
            });
        });
        
        // 返回 Promise（作为 Future）
        return promise;
    }

    /**
     * Demo 4: Promise 实现重试机制
     */
    private static void demo4_PromiseWithRetry() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 4: Promise 实现重试机制");
        log.info("└─────────────────────────────────────────\n");
        
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new SimpleHandler());
                    }
                });
            
            ChannelFuture connectFuture = b.connect("localhost", 9000).sync();
            Channel channel = connectFuture.channel();
            
            // 使用带重试的发送方法
            log.info("发送消息（最多重试 3 次）");
            ChannelFuture sendFuture = sendWithRetry(channel, "Retry test\n", 3);
            
            sendFuture.addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.info("✅ 最终发送成功");
                } else {
                    log.error("❌ 重试失败: {}", future.cause().getMessage());
                }
            });
            
            sendFuture.sync();
            
            Thread.sleep(500);
            channel.close().sync();
            
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * 带重试的发送方法
     */
    private static ChannelFuture sendWithRetry(Channel channel, String message, int maxRetries) {
        ChannelPromise promise = channel.newPromise();
        
        doSendWithRetry(channel, message, maxRetries, 0, promise);
        
        return promise;
    }

    private static void doSendWithRetry(Channel channel, String message, 
                                        int maxRetries, int attempt, ChannelPromise promise) {
        log.info("   [重试] 尝试 {}/{}", attempt + 1, maxRetries);
        
        channel.writeAndFlush(
            Unpooled.copiedBuffer(message, StandardCharsets.UTF_8)
        ).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                log.info("   [重试] 发送成功");
                promise.setSuccess();
            } else {
                if (attempt < maxRetries - 1) {
                    log.warn("   [重试] 发送失败，准备重试...");
                    // 延迟后重试
                    channel.eventLoop().schedule(() -> {
                        doSendWithRetry(channel, message, maxRetries, attempt + 1, promise);
                    }, 500, java.util.concurrent.TimeUnit.MILLISECONDS);
                } else {
                    log.error("   [重试] 达到最大重试次数，放弃");
                    promise.setFailure(future.cause());
                }
            }
        });
    }

    /**
     * Demo 5: try 方法的使用
     */
    private static void demo5_TryMethods() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 5: trySuccess 和 tryFailure");
        log.info("└─────────────────────────────────────────\n");
        
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            // 1. setSuccess 只能调用一次
            log.info("1. setSuccess() - 只能设置一次");
            Promise<String> promise1 = new DefaultPromise<>(group.next());
            
            promise1.setSuccess("第一次设置");
            log.info("   第一次设置成功");
            
            try {
                promise1.setSuccess("第二次设置");  // 会抛异常
            } catch (IllegalStateException e) {
                log.error("   ❌ 第二次设置失败: {}", e.getMessage());
            }
            
            // 2. trySuccess 可以多次调用
            log.info("\n2. trySuccess() - 可以安全地多次调用");
            Promise<String> promise2 = new DefaultPromise<>(group.next());
            
            boolean result1 = promise2.trySuccess("第一次设置");
            log.info("   第一次 trySuccess: {}", result1);
            
            boolean result2 = promise2.trySuccess("第二次设置");
            log.info("   第二次 trySuccess: {}", result2);  // 返回 false
            
            log.info("   最终结果: {}", promise2.getNow());
            
            // 3. 实际应用：多个线程竞争设置结果
            log.info("\n3. 多线程竞争设置结果");
            Promise<String> promise3 = new DefaultPromise<>(group.next());
            
            promise3.addListener(future -> {
                log.info("   ✅ 最终胜出的结果: {}", future.getNow());
            });
            
            // 启动 3 个线程竞争
            for (int i = 1; i <= 3; i++) {
                final int threadNum = i;
                new Thread(() -> {
                    try {
                        Thread.sleep((long) (Math.random() * 100));
                        boolean success = promise3.trySuccess("线程 " + threadNum);
                        log.info("   线程 {} trySuccess: {}", threadNum, success);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }).start();
            }
            
            promise3.sync();
            
            Thread.sleep(500);
            
        } finally {
            group.shutdownGracefully();
        }
    }

    // 简单的处理器
    @Slf4j
    static class SimpleHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            ByteBuf buf = (ByteBuf) msg;
            log.info("[客户端] 收到回复: {}", buf.toString(StandardCharsets.UTF_8).trim());
            buf.release();
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            log.error("[客户端] 异常", cause);
            ctx.close();
        }
    }
}

/*
=== 运行结果分析 ===

1. Promise 的本质：

   Promise = Future + 可写能力
   
   - Future 视角：读取结果（消费者）
   - Promise 视角：设置结果（生产者）

2. Promise 的核心方法：

   设置成功：
   - setSuccess(V result)  - 设置成功结果（只能调用一次）
   - trySuccess(V result)  - 尝试设置成功（可多次调用）
   
   设置失败：
   - setFailure(Throwable cause)  - 设置失败原因（只能调用一次）
   - tryFailure(Throwable cause)  - 尝试设置失败（可多次调用）

3. Promise 的使用场景：

   ✅ 封装异步操作
   ✅ 桥接不同的异步 API
   ✅ 实现重试机制
   ✅ 组合多个异步操作
   ✅ 超时控制

4. set vs try 方法的区别：

   setSuccess/setFailure：
   - 只能调用一次
   - 重复调用会抛出 IllegalStateException
   - 适合单线程环境

   trySuccess/tryFailure：
   - 可以多次调用
   - 返回 boolean 表示是否设置成功
   - 适合多线程竞争场景

5. ChannelPromise vs Promise：

   ChannelPromise：
   - 专门用于 Channel 操作
   - 通过 channel.newPromise() 创建
   - 绑定到特定的 Channel
   
   Promise：
   - 通用的 Promise
   - 可以用于任何异步操作
   - 需要传入 EventExecutor

=== 关键知识点 ===

1. Promise 是可写的 Future
2. 一个 Promise 只能设置一次结果
3. 使用 try 方法可以安全地多次调用
4. Promise 用于生产者，Future 用于消费者
5. Promise 可以在不同线程间传递异步结果

=== 最佳实践 ===

✅ 使用 Promise 封装异步操作
✅ 使用 try 方法避免重复设置异常
✅ 在 EventLoop 中设置 Promise 结果
✅ 为 Promise 添加监听器处理结果
✅ 使用 Promise 实现重试等高级功能

❌ 不要多次调用 setSuccess/setFailure
❌ 不要忘记设置 Promise 的结果
❌ 不要在 Promise 已完成后再设置
❌ 不要在错误的线程中设置结果
*/

